package com.smart.thrift.discoveryandbalance.discovery.zookeeper;

import com.smart.thrift.config.client.ThriftClientManager;
import com.smart.thrift.discoveryandbalance.discovery.ServerListUpdater;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.zookeeper.Watcher;

import java.util.Objects;

import static com.smart.thrift.discoveryandbalance.discovery.zookeeper.ZookeeperRegistry.ZOOKEEPER_NODE_PREFIX;

public class ThriftZookeeperServerListUpdater implements ServerListUpdater {

    private ZookeeperRegistry registry;
    private NodeChildrenChangeListener nodeChildrenChangeListener;
    private ConnectionStateChangeListener connectionStateChangeListener;

    public ThriftZookeeperServerListUpdater() {
    }

    @Override
    public void start(UpdateAction updateAction) {
        if (Objects.isNull(registry)){
            registry = (ZookeeperRegistry) ThriftClientManager.getInstance().getRegistry();
        }
        if(Objects.isNull(nodeChildrenChangeListener)){
            nodeChildrenChangeListener = new NodeChildrenChangeListener(updateAction);
        }
        if(Objects.isNull(connectionStateChangeListener)){
            connectionStateChangeListener = new ConnectionStateChangeListener(updateAction);
        }
        CuratorFramework zkClient = registry.getClient();
        zkClient.getCuratorListenable().addListener(nodeChildrenChangeListener);
        zkClient.getConnectionStateListenable().addListener(connectionStateChangeListener);
    }

    @Override
    public void stop() {
        if (Objects.isNull(registry)){
            registry = (ZookeeperRegistry) ThriftClientManager.getInstance().getRegistry();
        }
        CuratorFramework client = registry.getClient();
        if(Objects.isNull(nodeChildrenChangeListener)){
            client.getCuratorListenable().removeListener(nodeChildrenChangeListener);
        }
        if(Objects.isNull(connectionStateChangeListener)){
            client.getConnectionStateListenable().removeListener(connectionStateChangeListener);
        }

    }


    @Slf4j
    private static class NodeChildrenChangeListener implements CuratorListener {

        private final UpdateAction updateAction;

        public NodeChildrenChangeListener(UpdateAction updateAction) {
            this.updateAction = updateAction;
        }

        @Override
        public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
            //如果服务节点发生变化更新服务列表
            if (!Objects.isNull(event.getPath()) && event.getPath().startsWith(ZOOKEEPER_NODE_PREFIX) && event.getWatchedEvent() != null) {
                final Watcher.Event.EventType type = event.getWatchedEvent().getType();
                log.info("Zookeeper path {} event detected {}", event.getPath(), type);
                if (type == Watcher.Event.EventType.NodeChildrenChanged ||
                        type == Watcher.Event.EventType.NodeCreated ||
                        type == Watcher.Event.EventType.NodeDeleted) {
                    updateAction.doUpdate();
                }
            }
        }
    }

    @Slf4j
    private static class ConnectionStateChangeListener implements ConnectionStateListener {

        private final UpdateAction updateAction;

        public ConnectionStateChangeListener(UpdateAction updateAction) {
            this.updateAction = updateAction;
        }

        @Override
        public void stateChanged(CuratorFramework client, ConnectionState state) {
            if (state == ConnectionState.LOST) {
                while (true) {
                    try {
                        boolean connected = client.getZookeeperClient().blockUntilConnectedOrTimedOut();
                        if (connected) {
                            updateAction.doUpdate();
                            break;
                        }
                    } catch (InterruptedException e) {
                        log.warn("State changing monitoring interrupted", e);
                        break;
                    } catch (Exception e) {
                        e.printStackTrace();
                        log.error("Failed to find server on Zookeeper", e);
                    }
                }
            }
        }
    }
}
